package tk.xboot.flink.kfk.json;

import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.clients.producer.*;

import java.io.InputStream;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.Random;

public class KafkaJsonProducerMain {
    private static final ObjectMapper om = new ObjectMapper();//.registerModule(new JavaTimeModule());

    public static void main(String[] args) throws Exception {

        Properties props = new Properties();
        try (InputStream is = KafkaJsonProducerMain.class.getClassLoader().getResourceAsStream("kfk-producer.properties")) {
            props.load(is);
        }
        produceJsonMessage(props);
    }


    public static void produceJsonMessage(Properties props){
        try (Producer<String,String> producer = new KafkaProducer<>(props)) {
            int totalMessageCount = 10000;
            Random r = new Random();
            Map map = new HashMap<String, Object>();
            for (int i = 0; i < totalMessageCount; i++) {

                int m = Math.abs(r.nextInt()%(2<<2)) + 1;
                long cur = System.currentTimeMillis();
                final int x = i;

                map.clear();
                map.put("ts",cur);
                map.put("cnt", m);
                map.put("coin","coin"+m);
                String value = om.writeValueAsString(map);
                producer.send(new ProducerRecord<>("data-test",null,cur,cur+ "", value), new Callback() {
                    @Override
                    public void onCompletion(RecordMetadata metadata, Exception exception) {
                        if (exception != null) {
                            System.out.println("Failed to send message with exception " + exception);
                        } else {
                            System.out.println("Send data:"+"[idx:"+ x +" value:" + value + "]");
                        }
                    }
                });
                Thread.sleep(1000L);
            }
            //producer.close();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
    }
}
